📄 ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/metadata.py:268-269

The yield_pipeline_status method uses break when it encounters a run with start_time < cutoff_ts. This assumes that self.client.get_job_runs() returns runs in strictly descending chronological order (newest first).

If the Databricks API returns runs in a different order (e.g., ascending, or unordered), this break will cause the method to skip recent runs that appear after an older one in the list. This would result in incomplete pipeline status data being ingested.

Suggestion: Either:

1. Verify and document that get_job_runs() guarantees descending order (Databricks Jobs API does return runs in descending order by default when using list_runs)
2. Use continue instead of break to safely handle all runs regardless of ordering, though this loses the early-termination optimization
3. Add a sort before iterating: sorted(runs, key=lambda r: r.get('start_time', 0), reverse=True)

if run.start_time and run.start_time < cutoff_ts:
    continue  # safer: skip old runs without assuming order

📄 bootstrap/sql/migrations/native/1.11.8/mysql/schemaChanges.sql:32-36

The MySQL migration drops the timestamp column and re-adds it in a single ALTER TABLE statement. While MySQL can handle this atomically within one DDL statement, dropping a VIRTUAL generated column and re-adding it as STORED forces a full table rebuild on entity_extension_time_series. On large deployments, this table can contain millions of rows (pipeline statuses, test results, profiler data, etc.), which means:

1. Table lock duration: The ALTER TABLE will hold a metadata lock for the entire rebuild, blocking all reads/writes to this table.
2. Downtime risk: Any concurrent pipeline status writes or reads during migration will be blocked.

Consider using pt-online-schema-change or gh-ost for large deployments, or at minimum, document the expected downtime/lock duration in the migration comments so operators can plan accordingly. Also consider adding a note about the migration's impact in the PR description since this is a high-traffic table.

📄 openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/PipelineRepository.java:419

In addBulkPipelineStatus, the flow is:

1. Validate tasks (lines 389-396)
2. Bulk upsert to DB (line 405)
3. Bulk index to ES (line 407)
4. Store RDF (line 409)
5. Update pipeline entity index (line 419)

If step 3, 4, or 5 throws an exception, the DB records from step 2 have already been committed (JDBI useHandle auto-commits). Unlike the single-status path which wraps ES indexing in a try-catch (lines 360-365), the bulk path's bulkIndexPipelineExecutions already has its own try-catch internally, which is good. However, the searchRepository.updateEntityIndex(pipeline) call at line 419 is NOT wrapped in a try-catch. If it throws, the response will be an error even though the DB upsert succeeded, leaving the client uncertain about the operation's status.

Suggested fix: Wrap the entity index update in a try-catch to match the pattern used for ES execution indexing:

try {
    searchRepository.updateEntityIndex(pipeline);
} catch (Exception e) {
    LOG.error("Failed to update pipeline entity index in Elasticsearch", e);
}

📄 openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/PipelineRepository.java:419-423 📄 openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/PipelineRepository.java:419

The single-status addPipelineStatus method (line 371) calls reindexAcrossIndices("upstreamLineage.pipeline.fullyQualifiedName", ...) to update downstream lineage entities in the search index when a pipeline's status changes. The bulk path addBulkPipelineStatus (lines 419-423) only calls searchRepository.updateEntityIndex(pipeline) but omits the reindexAcrossIndices call entirely.

This means entities downstream of the pipeline (e.g., tables with lineage from this pipeline) won't have their search index updated when bulk statuses are ingested, causing stale upstreamLineage data in search results.

Suggested fix: Add the same reindexAcrossIndices call after searchRepository.updateEntityIndex(pipeline):

try {
    searchRepository.updateEntityIndex(pipeline);
    searchRepository.getSearchClient()
        .reindexAcrossIndices(
            "upstreamLineage.pipeline.fullyQualifiedName", pipeline.getEntityReference());
} catch (Exception e) {
    LOG.error("Failed to update pipeline entity index in Elasticsearch", e);
}

📄 openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/PipelineRepository.java:377-384

In addBulkPipelineStatus, when the list is null/empty, the method fetches the pipeline and returns it (lines 377-380). Then if the list is non-empty, it fetches the pipeline again (lines 383-384). This means in the normal (non-empty) path, the pipeline is fetched once, but the empty-list guard creates an unnecessary code path that returns ENTITY_UPDATED even though nothing was updated.

Consider either:

1. Returning Response.Status.NO_CONTENT for empty lists, or
2. Moving the pipeline fetch before the check and reusing it:

Pipeline pipeline = daoCollection.pipelineDAO().findEntityByName(fqn);
pipeline.setService(getContainer(pipeline.getId()));
if (pipelineStatuses == null || pipelineStatuses.isEmpty()) {
    return new RestUtil.PutResponse<>(Response.Status.OK, pipeline, ENTITY_NO_CHANGE);
}

📄 openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/PipelineRepository.java:395-400

The addBulkPipelineStatus method finds the latest status among the submitted batch and sets it as the pipeline's pipelineStatus. However, it doesn't check whether the pipeline already has a more recent status from a previous call. If someone submits a batch of older statuses after a newer one was already set, the pipeline's pipelineStatus would regress to an older timestamp.

For comparison, the existing single-status addPipelineStatus also doesn't guard against this, so this may be by-design. But with bulk operations, the risk is higher since a batch of historical data is a common use case.

Suggestion: Consider comparing the batch's latest timestamp against the pipeline's current pipelineStatus timestamp before overwriting.

...and 2 more resolved from earlier reviews